package com.yanggu.spark.core.demand.pageJump

import org.apache.spark.{SparkConf, SparkContext}

//需求3：页面单跳转换率统计
object PageJump_01 {

  def main(args: Array[String]): Unit = {

    //1. 创建sparkConf配置对象
    val sparkConf = new SparkConf().setAppName("PageJump").setMaster("local[*]")

    //2. 创建sparkContext上下文环境对象
    val sparkContext = new SparkContext(sparkConf)

    //3. 读取用户访问记录数据
    val fileRDD = sparkContext.textFile("input/user_visit_action.txt")

    //将数据缓存起来
    fileRDD.cache

    //计算分母
    val map = fileRDD
      .map(line => {
        val datas = line.split("_")
        (datas(3), 1)
      })
      .reduceByKey(_ + _)
      .collect
      .toMap

    //计算分子
    fileRDD
      .map(line => {
        val datas = line.split("_")
        //将数据转换成(sessionId, (时间, 页面id))
        (datas(2), (datas(0), datas(3)))
      })
      //根据key进行分组
      //(sessionId, (时间, 页面id)) => (sessionId, Iterable[时间, 页面id])
      .groupByKey
      //key不变, 对value进行运算
      .mapValues { datas =>
        val tuples = datas.toList
          //根据行为时间进行升序排序
          .sortBy(_._1)
          //这里只需要页面id
          .map(_._2)
        //这里使用拉链
        //1, 2, 3, 4
        //2, 3, 4
        tuples
          //(1, 2), (2, 3), (3, 4)
          .zip(tuples.tail)
          //((1, 2), 1), ((2, 3), 1), ((3, 4), 1)
          .map {
            case (pageId1, pageId2) => ((pageId1, pageId2), 1)
          }
      }
      //结构的转换, 这里sessionId不再需要, 只需要List[(pageId1, pageId2, 1)]
      .map(_._2)
      //扁平化操作, 这里将List[(pageId1, pageId2, 1)] => (pageId1, pageId2, 1)
      .flatMap(list => list)
      //wordCount操作 (pageId1, pageId2, 1) => (pageId1, pageId2, sum)
      .reduceByKey(_ + _)
      //计算页面跳转率
      //这里其实可以进行优化, 可以将map作为广播变量
      .foreach {
        case ((pageId1, pageId2), sum) =>
          val pageId1Sum = map.getOrElse(pageId1, 1)
          println(s"$pageId1 - $pageId2, 页面跳转率: ${sum.toDouble / pageId1Sum}")
      }
  }

}

